package com.didd.core.common.utils;

import com.didd.core.common.utils.jctools.MpmcUnboundedQueueExecutor;
import lombok.extern.slf4j.Slf4j;

import java.util.concurrent.ConcurrentHashMap;

/**
 * @author diddd
 * @Date 2020/12/17
 * @Note MPMC 高性队列 适合处理异步任务，无锁 +去线程池
 */
@Slf4j
public class AsyncEnhanceQueueUtils {
    public interface FutureListener<T> {
        void operationComplete(T t, String index, Throwable e) throws Exception;
    }

    public static AsyncEnhanceQueueProxy createAsyncEnhanceQueueProxy(AsyncEnhanceQueueProxyConfig asyncEnhanceQueueProxyConfig) {
        return new AsyncEnhanceQueueProxy(asyncEnhanceQueueProxyConfig);
    }

    public static class AsyncEnhanceQueueProxyConfig {
        private static final int DEFAULT_THREAD_MAX_CNT = 10;
        private int threadMaxCnt = DEFAULT_THREAD_MAX_CNT;
        private static final int DEFAULT_CHUNK_SIZE = 512;
        private int initialCapacity = DEFAULT_CHUNK_SIZE;
        private FutureListener futureListener;

        public AsyncEnhanceQueueProxyConfig setThreadMaxCnt(int threadMaxCnt) {
            this.threadMaxCnt = threadMaxCnt;
            return this;
        }

        public AsyncEnhanceQueueProxyConfig setInitialCapacity(int initialCapacity) {
            this.initialCapacity = initialCapacity;
            return this;
        }

        public AsyncEnhanceQueueProxyConfig setFutureListener(FutureListener futureListener) {
            this.futureListener = futureListener;
            return this;
        }

        public int getThreadMaxCnt() {
            return threadMaxCnt;
        }

        public int getInitialCapacity() {
            return initialCapacity;
        }
    }

    public static class AsyncEnhanceQueueProxy {
        private AsyncEnhanceQueueProxyConfig asyncEnhanceQueueProxyConfig;

        public AsyncEnhanceQueueProxy(AsyncEnhanceQueueProxyConfig asyncEnhanceQueueProxyConfig) {
            this.asyncEnhanceQueueProxyConfig = asyncEnhanceQueueProxyConfig;
        }

        private final ConcurrentHashMap<String, MpmcUnboundedQueueExecutor> mpmcUnboundedQueueExecutorMap = new ConcurrentHashMap();
        private static final int EXECUTOR_MAX_SIZE = 10;

        public MpmcUnboundedQueueExecutor getIndex(String index) {
            ObjectUtil.checkNotNull(index, "AsyncEnhanceQueueUtils 索引不为空");
            return mpmcUnboundedQueueExecutorMap.get(index);
        }

        public void computeIfAbsent(String index) {
            computeIfAbsent(index, asyncEnhanceQueueProxyConfig);
        }

        public void computeIfAbsent(String index, AsyncEnhanceQueueProxyConfig asyncEnhanceQueueProxyConfig) {
            if (mpmcUnboundedQueueExecutorMap.size() > EXECUTOR_MAX_SIZE) {
                throw new IllegalArgumentException("mpmcUnboundedQueueExecutorMap 元素过多超过阀值 =" + mpmcUnboundedQueueExecutorMap.size());
            }
            mpmcUnboundedQueueExecutorMap.computeIfAbsent(index, (c) -> {
                MpmcUnboundedQueueExecutor mpmcUnboundedQueueExecutor = MpmcUnboundedQueueExecutor.mpmcUnboundedQueueExecutorBuilder().setMinSize(asyncEnhanceQueueProxyConfig.getInitialCapacity())
                        .setNamedThreadFactory(index + "-AsyncEnhanceQueue").
                                setThreadMaxCnt(asyncEnhanceQueueProxyConfig.getThreadMaxCnt())
                        .build();
                return mpmcUnboundedQueueExecutor;
            });
        }
    }

}

